package com.example.springbootlearn.netty.handler;


import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.example.springbootlearn.netty.St;
import com.example.springbootlearn.netty.annotation.NettyHandler;
import com.example.springbootlearn.netty.dto.BizMsg;
import com.example.springbootlearn.netty.okhttp.TxAsrWebsocket;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.dromara.dynamictp.common.queue.VariableLinkedBlockingQueue;
import org.dromara.dynamictp.core.executor.OrderedDtpExecutor;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
@ChannelHandler.Sharable
@NettyHandler(value = 100)
public class BizMsgHandler extends SimpleChannelInboundHandler<BizMsg> {

    private static final ConcurrentLinkedQueue<Integer> availableIndices = new ConcurrentLinkedQueue<>();
    private static Map<String, Integer> phoneAndThreadIndexMap = new ConcurrentHashMap<>();
    private static Map<String, Long> timeStampMap = new HashMap<>();
    private static int INITIAL_MAX_THREAD_COUNT = 40;
    private static final AtomicInteger limitMaxThreadCount = new AtomicInteger(INITIAL_MAX_THREAD_COUNT);

    static {
        for (int i = 0; i < INITIAL_MAX_THREAD_COUNT; i++) {
            availableIndices.offer(i);
        }
    }

    private final AtomicInteger atomicInteger = new AtomicInteger(0);
    AttributeKey<Integer> threadIndexAttr = AttributeKey.valueOf("threadIndex");
    OrderedDtpExecutor executor = new OrderedDtpExecutor(INITIAL_MAX_THREAD_COUNT, 40, 1, TimeUnit.MINUTES, new VariableLinkedBlockingQueue<>(1024));

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, BizMsg bizMsg) throws InterruptedException {

        Channel channel = ctx.channel();
        Integer threadIndex = channel.attr(threadIndexAttr).get();
        if (threadIndex == null) {
            threadIndex = allocateThread(bizMsg.getMessageId());
            channel.attr(threadIndexAttr).set(threadIndex);
        }

        // 线程池处理业务逻辑
        executor.execute(() -> doBizMsgHandle(ctx, bizMsg), threadIndex);

    }

    public Integer allocateThread(String phoneNumber) {
        return phoneAndThreadIndexMap.computeIfAbsent(phoneNumber, k -> {
            Integer newIndex = availableIndices.poll();
            if (newIndex == null) {
                newIndex = limitMaxThreadCount.incrementAndGet();
                log.info("超出线程池最大线程数{}", newIndex);
            }
            return newIndex;
        });
    }

    //    public synchronized void releaseThread(String phoneNumber) {
//        Integer theadIndex = phoneAndThreadindexMap.remove(phoneNumber);
//        Optional.ofNullable(theadIndex).ifPresent(availableIndices::add);
//        log.info("释放线程索引：{}，当前线程索引数剩余：{}", theadIndex, availableIndices.size());
//    }
    public void releaseThread(String phoneNumber) {
        Integer threadIndex = phoneAndThreadIndexMap.remove(phoneNumber);
        if (threadIndex != null) {
            if (threadIndex < INITIAL_MAX_THREAD_COUNT) {
                availableIndices.offer(threadIndex);
            }
            log.info("释放线程索引：{}，当前线程索引数剩余：{}", threadIndex, availableIndices.size());
        }
    }

    private void doBizMsgHandle(ChannelHandlerContext ctx, BizMsg bizMsg) {
        TxAsrWebsocket txAsrWebsocket = SpringUtil.getBean(TxAsrWebsocket.class);
        String code = bizMsg.getCode();
        String messageId = bizMsg.getMessageId();
        String sessionId = messageId + "_" + bizMsg.getRole();
        String printMsg = messageId + "_" + bizMsg.getMsg();
        if (Objects.equals(code, "invite")) {
            atomicInteger.incrementAndGet();
            timeStampMap.put(messageId, System.currentTimeMillis());
            log.info("线程名字：{}，收到invite消息：{}", Thread.currentThread().getName(), printMsg);
            txAsrWebsocket.initAsrWsClient(messageId);
            St.put(messageId, ctx.channel());
            ThreadUtil.safeSleep(RandomUtil.randomInt(200, 500));
        } else if (Objects.equals(code, "ack")) {
            log.info("线程名字：{}，收到ack消息：{}", Thread.currentThread().getName(), bizMsg);
            txAsrWebsocket.sendData(sessionId, JSONUtil.toJsonStr(bizMsg));
        } else if (Objects.equals(code, "bye")) {
            log.info("线程名字：{}，收到bye消息：{}", Thread.currentThread().getName(), printMsg);
            releaseThread(messageId);
            txAsrWebsocket.closeAsrWsClient(messageId);
            St.remove(messageId);
            Long startTime = timeStampMap.remove(messageId);
            log.info("线程名字：{}，messageIdxxxx {} 并发连接{}个 处理时间：{}ms", Thread.currentThread().getName(), messageId, atomicInteger.get(), System.currentTimeMillis() - startTime);
            ThreadUtil.safeSleep(RandomUtil.randomInt(190, 300));
        }
    }
}